/*
开发目标：将windows下的processor搬运到linux上，练习使用mysql，linux多线程
具体细节：
1.提高并发能力，多少合适？
2.提高程序运行速度，计算没条数据的处理速度
3.完整的数据记录
4.使用redis缓存实时数据
5.日志打印到syslog
*/
#include <string>
#include <iostream>
#include <syslog.h>
#include <sys/types.h>
#include <unistd.h>
#include <boost/format.hpp>
#include "../include/global.h"
#include "../include/mysql/CMySqlOp.h"
#include "../include/config/CConfigIni.h"
#include "../include/machine/CMachineGroup.h"
#include <sys/socket.h>
#include <sys/select.h>
#include <sys/errno.h>
#include <sys/time.h>
#include <arpa/inet.h>      //struct sockaddr_in
#include <arpa/inet.h>
#include <thread>
#include "../../include/thread/threadHandle.h"
//include "boost/asio.hpp"

CMachineGroup gMachGroup;           //全部机器信息

//回包队列
std::queue<DATA_SEND> gQueDataSend;
std::mutex gmtDataSend;
std::condition_variable gcvDataSend;  

#define USE_SELECT  0     //使用select模型
#define USE_POLL     1   //使用poll模型
#define USE_EPOLL     2  //使用epoll模型

#define TCP_MODE    USE_SELECT

const int BACKLOG = 10;                 //完成三次握手但没有accept的队列的长度
const int MAX_CLIENT_SIZE = 2000;       //客户端能同时处理的最大连接数
const int MAX_RECV_BUF = 1025;          //最多接受1024个字符
const int MAX_SEND_BUF = 1025;          //最大发送字符串 

using namespace std;
using boost::format;

//将进程变成守护进程
void DaemonProcess();

int main(int argc, char** argv)
{    
    format fmt ;

    //守护进程
    //DaemonProcess();
 
    //启动日志系统
    openlog("Processor",LOG_CONS|LOG_PID,LOG_LOCAL3);

    //读取配置文件
    string iniPath = CConfigIni::APP_PATH+"/config.ini";
    if(!G_INIFILE->LoadConfigIni(iniPath))
    {
        fmt = boost::format("配置文件加载失败,%s,搜索目录[%s]");
        fmt%G_INIFILE->GetErrMsg();
        fmt%iniPath;
        syslog(LOG_ERR,(fmt.str()).c_str());        
        closelog();         //关闭日志
        return -1;
    }
    else
    {
        syslog(LOG_ERR,"配置文件加载成功");        
    }

    //测试数据库连接性
    std::string errMsg = "";
    if(!CMySQLOper::TestMysql(G_INIFILE->dbHost,G_INIFILE->dbPort,
    G_INIFILE->dbUser,G_INIFILE->dbPwd,G_INIFILE->dbSchema,errMsg))
    {
        format fmt = boost::format("数据库测试失败,%s");
        fmt%errMsg;
        syslog(LOG_ERR,(fmt.str()).c_str());
        closelog();
        return -1;
    }

    //加载数据库信息
    CMySQLOper mysql;
    if(mysql.Connect(G_INIFILE->dbHost,G_INIFILE->dbPort,
    G_INIFILE->dbUser,G_INIFILE->dbPwd,G_INIFILE->dbSchema))
    {
        //加载设备信息
        syslog(LOG_INFO,"加载设备信息");
        if(!gMachGroup.LoadMachine(&mysql))
        {
            fmt = boost::format("设备信息加载失败，%s");
            fmt%gMachGroup.GetErrMsg().c_str();
            syslog(LOG_ERR,fmt.str().c_str());
            closelog();
            mysql.CloseConnect();
            return -1;
        }
        else
        {
            //打印设备信息到日志            
            int totalMach = gMachGroup.GetMachineSize();
            if(totalMach <= 0)
            {
                syslog(LOG_ERR,"未查询到设备信息");
                closelog();
                mysql.CloseConnect();
                return -1;
            }
            else
            {
                for(unsigned int i = 0; i< totalMach;i++)
                {
                    CMachine *pMach = gMachGroup.GetMachineByIndex(i);
                    if(pMach != NULL)
                    {
                        fmt = boost::format("机器ID[%d],机器名[%s],机器类别[%s],采集器ID[%d]");
                        fmt% pMach->machID;
                        fmt% pMach->machName;
                        fmt% pMach->typeName;
                        fmt% pMach->collectorID;
                        syslog(LOG_INFO,fmt.str().c_str());
                    }
                }
            }            
        }

        //恢复上一次的数据状态
        syslog(LOG_INFO,"恢复设备状态");
        if(!gMachGroup.RecoverMachineData(&mysql))
        {
            fmt = boost::format("设备状态恢复失败，%s");
            fmt%gMachGroup.GetErrMsg().c_str();
            syslog(LOG_ERR,fmt.str().c_str());
            closelog();
            mysql.CloseConnect();
            return -1;
        }

    }
    else
    {
        fmt = boost::format("数据库连接失败，%s");
        fmt%gMachGroup.GetErrMsg().c_str();
        syslog(LOG_ERR,fmt.str().c_str());
        closelog();
        return -1;
    }    

    //创建数据回发线程
    std::thread threadSend = std::thread(ThreadSendData);
    threadSend.detach();

    //创建服务器，响应客户端连接I/O复用（同步、非阻塞）
    syslog(LOG_INFO,"启动socket服务，响应客户端请求");

#if TCP_MODE == USE_SELECT      //使用select模型（同步、非阻塞）    
    char recvBuf[MAX_RECV_BUF] = {0x00};
    bzero(recvBuf,0x00);
    struct sockaddr_in sockAddr;
    memset(&sockAddr, 0,sizeof(sockAddr));
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_port = htons(G_INIFILE->tcpPort);        //端口号
    sockAddr.sin_addr.s_addr = htonl(INADDR_ANY); //任意IP可连接

    //创建socket
    int listen_fd = socket(AF_INET,SOCK_STREAM,0);
    if(listen_fd < 0)
    {
        syslog(LOG_ERR,"套接字创建失败");
        return 0;
    }

    //绑定套接字
    int ret = bind(listen_fd,(struct sockaddr*)&sockAddr,sizeof(sockAddr));
    if(ret < 0)
    {
        syslog(LOG_ERR,"套接字绑定失败");
        return 0;
    }

    //启动监听
    ret = listen(listen_fd,BACKLOG);          //允许BACKLOG个连接未被处理
    if(ret < 0)
    {
        syslog(LOG_ERR,"套接字监听启动失败");
        return 0;
    }

    fd_set server_fd_set;       //监听集合
    int max_fd = -1;            //fd集合的最大值
    struct timeval tv;          //超时时间设置
    int client_fds[MAX_CLIENT_SIZE] = {0};

    while(true)
    {
        tv.tv_sec = 0;
        tv.tv_usec = 500;

        //每次循环都要清空集合，否则不能检测描述符变化 
        FD_ZERO(&server_fd_set); 
        //FD_SET(STDIN_FILEON,&server_fd_set); //标准输入进集合

        //服务器socket添加进集合
        FD_SET(listen_fd,&server_fd_set);   
        if(max_fd < listen_fd)
            max_fd = listen_fd;

        //已经连接的客户端socket添加进集合，此处可改进
        for(int i = 0; i< MAX_CLIENT_SIZE;i++)
        {
            if(client_fds[i] != 0)
            {
                FD_SET(client_fds[i],&server_fd_set);
                if(max_fd < client_fds[i])
                    max_fd = client_fds[i];
            }
        }

        //监听队列，tv=NULL时阻塞，=0时没有等待超时，>0时等待超时
        int ret = select(max_fd+1,&server_fd_set,NULL,NULL,&tv);
        if(ret < 0)
        {
            syslog(LOG_ERR,"select错误，等待下次轮询");
            continue;
        }
        else if( ret == 0)
        {
            //select等待超时            
            continue;
        }
        else
        {//有响应
            //判断是否是新连接到来
            if(FD_ISSET(listen_fd, &server_fd_set))
            {
                //新的连接请求到来
                struct sockaddr_in client_address;
            	socklen_t address_len;
            	int client_sock_fd = accept(listen_fd, (struct sockaddr *)&client_address, &address_len);
        	    if(client_sock_fd > 0)
                {
                    int index = -1;
                    for(int i = 0;i< MAX_CLIENT_SIZE;i++)
                    {
                        if(client_fds[i] == 0)
        			    {
        			    	index = i;
        			    	client_fds[i] = client_sock_fd;
        			    	break;
        			    }
                    }

                    //新连接到来时的提示
                    char temp[128] = {0x00};
                    if(index >= 0)
                    {                           
                        sprintf(temp,"新客户端[%s:%d(%d)]响应成功，剩余空闲连接数[%d]",
                        inet_ntoa(client_address.sin_addr), ntohs(client_address.sin_port),
                        client_sock_fd,MAX_CLIENT_SIZE-index-1);
                        syslog(LOG_INFO,temp);
                    }                     
        		    else
        			{
                        sprintf(temp,"新客户端[%s:%d]响应成功，但连接队列以满，强制关闭",
                        inet_ntoa(client_address.sin_addr), ntohs(client_address.sin_port));
                        syslog(LOG_ERR,temp);        				            				    
        			}   
                }                       			        		
            }

            //判断是否是已经连接的句柄消息
            for(int i = 0;i< MAX_CLIENT_SIZE;i++)
            {
                //过滤有效链接句柄
                if(client_fds[i] != 0)
                {
                    //测试是否是响应的句柄
                    if(FD_ISSET(client_fds[i],&server_fd_set))
                    {
                        int len = recv(client_fds[i],recvBuf,MAX_RECV_BUF,0);
                        if(len > 0)
                        {
                            //强制截断，防止崩溃
                            int tempLen = len > MAX_RECV_BUF - 1?MAX_RECV_BUF-1:len;                            
                            recvBuf[tempLen] = 0x00;   

                            char *pTemp = new char[tempLen+100];
                            bzero(pTemp,tempLen+100);
                            sprintf(pTemp,"客户端(%d)接收数据：%s",client_fds[i],recvBuf);
                            syslog(LOG_INFO,pTemp);
                            delete []pTemp;                         

                            //创建数据处理线程(或者将数据发送到消息中间件，通知其它线程处理)
                            std::thread taskThread = std::thread(ThreadTaskHanlde,client_fds[i],recvBuf,len);
                            taskThread.detach();                            
                        }
                        else if(len < 0)
                        {
                            //接收出错
                            char temp[128];
                            bzero(temp,128);                            
                            sprintf(temp,"客户端(%d)数据接收错误",client_fds[i]);
                        }
                        else
                        {
                            //断开连接
                            char temp[128];
                            bzero(temp,128);                            
                            sprintf(temp,"客户端(%d)断开连接",client_fds[i]);
                            client_fds[i] = 0;
                            syslog(LOG_INFO,temp);     
                        }
                    }
                }
            }
        }
    }

#elif TCP_MODE == USE_POLL      //使用poll模型
    int a = 0; 
    a = a;
#elif TCP_MODE == USE_EPOLL     //使用epoll模型

    int a = 0;
    a = a;
#else
    int a = a;
    a = a;
#endif              //同步阻塞服务器    
    return 0;
}


//将进程变成守护进程
void DaemonProcess()
{
    return;
}
